package com.gientech.Controller;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.UnsupportedEncodingException;
import java.util.List;

@RestController
@RequestMapping("/rocketmq/consumer")
public class ConsumerController {


    @Value("consume1")
    private String consumerGroup;
    @GetMapping("/consume")
    public void consume(){
        //消费者的组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer");
        //指定NameServer地址，多个地址以 ; 隔开
        consumer.setNamesrvAddr("127.0.0.1:9876");
        try {
            //设置consumer所订阅的Topic和Tag，*代表全部的Tag
            //consumer.subscribe("testOrderTopic", "*");
            consumer.subscribe("TopicTest", "*");

            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    System.out.printf("%s Receive New Message: %s \n ",Thread.currentThread().getName(),msgs);
                    try {
                        String s = new String(msgs.get(0).getBody(), "UTF-8");

                        System.out.printf("New Message: %s \n ",s);
                    } catch (UnsupportedEncodingException e) {
                        throw new RuntimeException(e);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            System.out.println("consumer start");
            //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，跳过历史消息
            //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
//            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//            //MessageListenerOrderly 这个是有序的
//            //MessageListenerConcurrently 这个是无序的,并行的方式处理，效率高很多
//            consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
//                try {
//                    for (MessageExt messageExt : list) {
//                        System.out.println("messageExt: " + messageExt);//输出消息内容
//                        String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
//                        System.out.println("消费响应：msgId : " + messageExt.getMsgId() + ",  msgBody : " + messageBody);//输出消息内容
//                    }
//                } catch (Exception e) {
//                    e.printStackTrace();
//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
//                }
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
//            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            System.out.println("done.-------------------");
        }
    }
}
